package com.allwe.client.concurrent;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 线程安全消费者 - 测试demo
 *
 * @Author: AllWe
 * @Date: 2024/09/27/12:19
 */
@Data
@Slf4j
public class ConcurrentConsumer implements Runnable {
    /**
     * 消费者配置参数
     */
    private Properties properties;

    /**
     * 群组id
     */
    private String groupId;

    /**
     * 消费主题
     */
    private String topicName;

    /**
     * 消费者实例
     */
    private KafkaConsumer<String, String> consumer;

    public ConcurrentConsumer(Properties properties, String groupId, String topicName) {
        this.properties = properties;
        this.groupId = groupId;
        this.topicName = topicName;

        // 补充配置参数
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 创建消费者实例 - 每一个线程都创建自己的消费者,避免共享相同的消费者实例
        consumer = new KafkaConsumer<>(properties);
        // 配置消费主题
        consumer.subscribe(Collections.singleton(topicName));
    }

    @Override
    public void run() {
        try {
            String threadName = Thread.currentThread().getName();
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> record : records) {
                    StringBuilder stringBuilder = new StringBuilder(threadName).append("|-");
                    stringBuilder.append("partition:").append(record.partition());
                    stringBuilder.append("offset:").append(record.offset());
                    stringBuilder.append("key:").append(record.key());
                    stringBuilder.append("value:").append(record.value());

                    System.out.println(stringBuilder);
                }
            }
        } finally {
            consumer.close();
        }
    }
}
